/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.rocketmq;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.FailureMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import cn.ac.iie.di.global.GlobalParas;
//import com.alibaba.rocketmq.common.message.Message;
//import cn.ac.iie.di.wx.picmover.kill.KillHandler;
//import cn.ac.iie.di.wx.picmover.util.GlobalParameters;
import java.util.logging.Level;
import org.apache.log4j.Logger;

/**
 *
 * @author YG
 */
public class RocketMqClient implements Runnable {

    public static Logger log = Logger.getLogger(RocketMqClient.class.getName());
    public String oriMqString = "";
    public static String oriMq = "";
    public static String group = "";

    private REConnection consuConn = new REConnection(GlobalParas.IP_PORT);
    private static REReceiveSession session;

    public RocketMqClient(String oriMqString) {
        this.oriMq = oriMqString.split("[:]")[0];
        this.group = oriMqString.split("[:]")[1];
//        killHandler.setConn(consuConn);
    }

    @Override
    public void run() {
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定
//        REConnection conn = new REConnection(GlobalParameters.nameserver);
        //生成builder，参数为topic，由具体业务决定
        REReceiveSessionBuilder builder = null;
        try {
            builder = (REReceiveSessionBuilder) consuConn.getReceiveSessionBuilder(this.oriMq);
        } catch (REConnectionException ex) {
            log.error(ex, ex);
        }
        //设置receiver 的两个处理函数：
        //一个是正常数据处理，实现REAbstractReceiveMessageHandler<REMessage>接口。返回true则成功，否则会发给下一个handler。
        if (builder == null) {
            log.error("init receiver fail , will exit!");
            System.exit(0);
        }
        builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
            @Override
            public boolean handle(REMessage message) {

                Message msg = new Message(message);
                log.info("get " + msg.getData().size() + " message from " + oriMq);
//                log.info("列名是:"+msg.getColNames());
//                log.info("类型是:"+msg.getTypes());
//                log.info("值是:"+msg.getData());
                try {
                    GlobalParas.PULL_MQ_QUEUE.put(msg);
                } catch (InterruptedException ex) {
                    log.error(ex, ex);
                }
                return true;
            }
        });
        //另一个是错误数据处理，实现REAbstractReceiveMessageHandler<byte[]>接口。返回true则成功，否则会退回去，重新处理。
        builder.setFailureHandler(new REAbstractReceiveMessageHandler<FailureMessage>() {
            @Override
            public boolean handle(FailureMessage msg) {
                throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
            }
        });
        //设置组名。相同的组名处理的是同一份数据，不同的组名处理的是不同的数据备份。
        builder.setGroupName(this.group);
        //设置从处理起始位置。如果设置的是CONSUME_FROM_TIMESTAMP，需要同时设置起始时间戳
//        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
        builder.setConsumeTimestamp(GlobalParas.GET_MQ_TIME);
        //设置消费线程数
        builder.setConsumeThreadNum(4);
        //生成session
        try {
            session = (REReceiveSession) builder.build();
            session.start();
        } catch (REConnectionException ex) {
        }
    }

    public static void shutDownConsumer() throws InterruptedException {
        if (session != null) {
            try {
                session.shutdown();
            } catch (REConnectionException ex) {
                java.util.logging.Logger.getLogger(RocketMqClient.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
//        while (!GlobalParameters.picHandlerQueue.isEmpty()) {
//            Thread.sleep(500);
//        }
    }

}
